package com.insight.event;

import com.insight.common.enums.MetadataChangeEventType;
import com.insight.dto.event.MetadataChangeEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;

/**
 * T076: Kafka元数据事件发布器
 * 负责将元数据变更事件发布到Kafka主题
 *
 * @author Insight Team
 */
@Slf4j
@Component
@ConditionalOnBean(KafkaTemplate.class)
@RequiredArgsConstructor
public class MetadataEventPublisher {

    private final KafkaTemplate<String, MetadataChangeEvent> kafkaTemplate;

    /**
     * Kafka主题名称
     */
    private static final String TOPIC = "metadata.changes";

    /**
     * 发布表结构变更事件
     *
     * @param tableId   表ID
     * @param tableName 表名
     */
    public void publishTableStructureChanged(Long tableId, String tableName) {
        log.info("发布表结构变更事件: tableId={}, tableName={}", tableId, tableName);

        MetadataChangeEvent event = MetadataChangeEvent.builder()
                .eventType(MetadataChangeEventType.TABLE_STRUCTURE_CHANGED)
                .tableId(tableId)
                .tableName(tableName)
                .timestamp(LocalDateTime.now())
                .build();

        sendEvent(event);
    }

    /**
     * 发布表添加事件
     *
     * @param datasourceId 数据源ID
     * @param tableId      表ID
     * @param tableName    表名
     */
    public void publishTableAdded(Long datasourceId, Long tableId, String tableName) {
        log.info("发布表添加事件: datasourceId={}, tableId={}, tableName={}",
                datasourceId, tableId, tableName);

        MetadataChangeEvent event = MetadataChangeEvent.builder()
                .eventType(MetadataChangeEventType.TABLE_ADDED)
                .datasourceId(datasourceId)
                .tableId(tableId)
                .tableName(tableName)
                .timestamp(LocalDateTime.now())
                .build();

        sendEvent(event);
    }

    /**
     * 发布表删除事件
     *
     * @param datasourceId 数据源ID
     * @param tableName    表名
     */
    public void publishTableRemoved(Long datasourceId, String tableName) {
        log.info("发布表删除事件: datasourceId={}, tableName={}", datasourceId, tableName);

        MetadataChangeEvent event = MetadataChangeEvent.builder()
                .eventType(MetadataChangeEventType.TABLE_REMOVED)
                .datasourceId(datasourceId)
                .tableName(tableName)
                .timestamp(LocalDateTime.now())
                .build();

        sendEvent(event);
    }

    /**
     * 发布采集任务完成事件
     *
     * @param taskId          任务ID
     * @param datasourceId    数据源ID
     * @param tablesCollected 采集的表数量
     */
    public void publishCollectionCompleted(Long taskId, Long datasourceId, Integer tablesCollected) {
        log.info("发布采集完成事件: taskId={}, datasourceId={}, tablesCollected={}",
                taskId, datasourceId, tablesCollected);

        Map<String, String> metadata = new HashMap<>();
        metadata.put("taskId", taskId.toString());
        metadata.put("tablesCollected", tablesCollected.toString());

        MetadataChangeEvent event = MetadataChangeEvent.builder()
                .eventType(MetadataChangeEventType.COLLECTION_COMPLETED)
                .datasourceId(datasourceId)
                .timestamp(LocalDateTime.now())
                .metadata(metadata)
                .build();

        sendEvent(event);
    }

    /**
     * 发布采集任务失败事件
     *
     * @param taskId       任务ID
     * @param datasourceId 数据源ID
     * @param errorMessage 错误信息
     */
    public void publishCollectionFailed(Long taskId, Long datasourceId, String errorMessage) {
        log.error("发布采集失败事件: taskId={}, datasourceId={}, error={}",
                taskId, datasourceId, errorMessage);

        Map<String, String> metadata = new HashMap<>();
        metadata.put("taskId", taskId.toString());
        metadata.put("errorMessage", errorMessage);

        MetadataChangeEvent event = MetadataChangeEvent.builder()
                .eventType(MetadataChangeEventType.COLLECTION_FAILED)
                .datasourceId(datasourceId)
                .timestamp(LocalDateTime.now())
                .metadata(metadata)
                .build();

        sendEvent(event);
    }

    /**
     * 发送事件到Kafka
     *
     * @param event 元数据变更事件
     */
    private void sendEvent(MetadataChangeEvent event) {
        kafkaTemplate.send(TOPIC, event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("发送Kafka事件失败: topic={}, event={}", TOPIC, event, ex);
                    } else {
                        log.debug("Kafka事件发送成功: topic={}, offset={}", TOPIC,
                                result.getRecordMetadata().offset());
                    }
                });
    }
}
